package com.surfilter.es;

import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Logger;

public class ElasticSearchProvider {
    private static Logger logger = Logger.getLogger(ElasticSearchProvider.class.getName());
    public static Map<String, TransportClient> client_map = new HashMap();
    private String client_key;
    private TransportClient client;

    public ElasticSearchProvider() {
    }

    public TransportClient getSearchClient() throws UnknownHostException {
        return this.getSearchClient("my-application", "47.94.88.80", true, 9300);
    }

    public TransportClient getSearchClient(String cluster_name, String link_addr) throws UnknownHostException {
        return this.getSearchClient(cluster_name, link_addr, true, 9300);
    }

    public TransportClient getSearchClient(String cluster_name, String link_addr, boolean is_sniff) throws UnknownHostException {
        return this.getSearchClient(cluster_name, link_addr, is_sniff, 9300);
    }

    public TransportClient getSearchClient(String cluster_name, String link_addr, boolean is_sniff, int port) throws UnknownHostException {
        if(!StringUtils.isBlank(cluster_name) && !StringUtils.isBlank(link_addr)) {
            if(client_map.containsKey(cluster_name + link_addr)) {
                return (TransportClient)client_map.get(cluster_name + link_addr);
            } else {
                Settings settings = Settings.builder().put("cluster.name", cluster_name).put("client.transport.sniff", is_sniff).build();
                PreBuiltTransportClient client = new PreBuiltTransportClient(settings, new Class[0]);
                client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(link_addr), port));
                client_map.put(cluster_name + link_addr, client);
                this.client_key = cluster_name + link_addr;
                this.client = client;
                return client;
            }
        } else {
            return this.getSearchClient();
        }
    }

    public void closeSearchClient() {
        if(null != this.client) {
            this.client.close();
            client_map.remove(this.client_key);
        }

    }

    public boolean isIndexExist(String index) throws UnknownHostException {
        TransportClient client = null == this.client?this.getSearchClient():this.client;
        IndicesExistsResponse inExistsResponse = (IndicesExistsResponse)client.admin().indices().exists(new IndicesExistsRequest(new String[]{index})).actionGet();
        if(inExistsResponse.isExists()) {
            logger.info("Index [" + index + "] is exist!");
        } else {
            logger.info("Index [" + index + "] is not exist!");
        }

        return inExistsResponse.isExists();
    }

    public boolean deleteIndex(String index) throws UnknownHostException {
        if(!this.isIndexExist(index)) {
            logger.info("Index is not exits!");
        }

        TransportClient client = null == this.client?this.getSearchClient():this.client;
        DeleteIndexResponse dResponse = (DeleteIndexResponse)client.admin().indices().prepareDelete(new String[]{index}).execute().actionGet();
        if(dResponse.isAcknowledged()) {
            logger.info("delete index " + index + "  successfully!");
        } else {
            logger.info("Fail to delete index " + index);
        }

        return dResponse.isAcknowledged();
    }

    public Map<String, Object> getSourceById(String index, String type, String id, String fields) {
        try {
            TransportClient e = null == this.client?this.getSearchClient():this.client;
            GetResponse getResponse = (GetResponse)e.prepareGet(index, type, id).setFetchSource(fields.split(","), (String[])null).get();
            logger.info(getResponse.getSourceAsString());
            return getResponse.getSource();
        } catch (Exception var7) {
            var7.printStackTrace();
            return null;
        }
    }

    public Map<String, Object> save(String index, String type, String id, Map<String, Object> data) throws UnknownHostException {
        TransportClient client = null == this.client?this.getSearchClient():this.client;
        IndexRequestBuilder builder = client.prepareIndex(index, type, id);
        IndexResponse response = (IndexResponse)builder.setSource(data).execute().actionGet();
        logger.info("save index:=>index:{" + index + "}, type:{" + type + "}, id:{" + id + "}, data:{" + data + "}, rsp:{" + response + "}");
        return data;
    }

    public int update(String index, String type, String id, Map<String, Object> data) throws UnknownHostException {
        int i = 2;

        do {
            try {
                if(this._innerUpdate(index, type, id, data)) {
                    return 1;
                }
            } catch (VersionConflictEngineException var7) {
                logger.warning(String.format("update index:=>index:%s, type:%s, id:%s, data:%s, rsp:%s", new Object[]{index, type, id, data, var7.getMessage()}));
            }
        } while(i-- > 0);

        return this._innerUpdate(index, type, id, data)?1:0;
    }

    public Map<String, Object> get(String index, String type, String id) throws UnknownHostException {
        TransportClient client = null == this.client?this.getSearchClient():this.client;
        GetRequestBuilder builder = client.prepareGet(index, type, id);
        GetResponse response = (GetResponse)builder.execute().actionGet();
        return response.isExists()?response.getSource():null;
    }

    private boolean _innerUpdate(String index, String type, String id, Map<String, Object> data) throws UnknownHostException {
        TransportClient client = null == this.client?this.getSearchClient():this.client;
        GetRequestBuilder getRequestBuilder = client.prepareGet(index, type, id);
        GetResponse getResponse = (GetResponse)getRequestBuilder.execute().actionGet();
        if(getResponse.isExists()) {
            long version = getResponse.getVersion();
            Map source = getResponse.getSource();
            source.putAll(data);
            IndexRequestBuilder builder = client.prepareIndex(index, type, id);
            IndexResponse response = (IndexResponse)builder.setVersion(version).setSource(source).execute().actionGet();
            logger.info("update index:=>index:{" + index + "}, type:{" + type + "}, id:{" + id + "}, data:{" + data + "}, rsp:{" + response + "}");
            return true;
        } else {
            throw new RuntimeException(String.format("can not get document:=>index:%s, type:%s, id:%s ", new Object[]{index, type, id}));
        }
    }

    public boolean bulkInsert(String index, String type, List data) {
        if(null != data && data.size() > 0) {
            BulkRequestBuilder bulkRequest = this.client.prepareBulk();

            for(int i = 0; i < data.size(); ++i) {
                bulkRequest.add(this.client.prepareIndex(index, type).setSource(new Object[]{data.get(i)}));
                if(i % 1000 == 0) {
                    bulkRequest.execute().actionGet();
                    logger.info("save 1000 datas index:=>index:{" + index + "}, type:{" + type + "}");
                }
            }

            bulkRequest.execute().actionGet();
            return true;
        } else {
            return false;
        }
    }

    /**
     * 批量导入数据
     *
     * @param index
     * @param type
     * @param unique
     * @param data
     * @return
     */
    public boolean bulkInsert4Unique(String index, String type, String unique, List data) {
//        String json = null;
//        int count = 0;
        //开启批量插入
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        for (int i = 0; i < data.size(); i++) {
            Map map = (Map) data.get(i);
            bulkRequest.add(client.prepareIndex(index, type, map.get(unique).toString()).setSource(map));
            //每一千条提交一次
            if (i % 1000 == 0) {
                bulkRequest.execute().actionGet();
                logger.info("save 1000 datas index:=>index:{" + index + "}, type:{" + type + "}");
            }
        }
        bulkRequest.execute().actionGet();
        return true;
    }

    public void putMapping(String indexType, XContentBuilder mapping, String indexName) {
        CreateIndexRequestBuilder cib = this.client.admin().indices().prepareCreate(indexName);
        cib.addMapping(indexType, mapping);
        cib.execute().actionGet();
    }
}
